package com.smile.frame.mq.kafka.controller;

import com.smile.frame.mq.kafka.config.KafkaConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author LGC
 */
@Slf4j
@Component
public class TestConsumerListener {

    /**
     * 手动提交
     *
     * @KafkaListener(groupId = "testGroup", topicPartitions = {
     * @TopicPartition(topic = "topic1", partitions = {"0", "1"}),
     * @TopicPartition(topic = "topic2", partitions = "0", partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100")
     * },
     * concurrency = "6")
     * <p>
     * concurrency就是同组下的消费者个数，就是并发消费数，必须小于等于分区总数
     */
    @KafkaListener(id = "testListener", groupId = KafkaConfig.GROUP_ID,
            topics = KafkaConfig.TEST_TOPIC, concurrency = "1",
            containerFactory = "kafkaManualAckListenerContainerFactory")
    public void testListener(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
        try {
            log.info("topic.test1 receive : \n" +
                    "data : " + record.value() + "\n" +
                    "key : " + record.key() + "\n" +
                    "topic : " + record.topic() + "\n" +
                    "partitionId : " + record.partition() + "\n" +
                    "offset : " + record.offset() + "\n" +
                    "timestamp : " + record.timestamp() + "\n"
            );
            // 手动提交
            acknowledgment.acknowledge();
        } catch (DuplicateKeyException duplicateKeyException) {
            // 处理异常情况
            log.error("数据重复", duplicateKeyException);
            // 重复数据，忽略掉，同步提交
            acknowledgment.acknowledge();
        }

    }


    /**
     * 手动提交 批量消费
     */
    @KafkaListener(id = "test1Listener", groupId = KafkaConfig.GROUP_ID,
            topics = KafkaConfig.TEST1_TOPIC, concurrency = "1",
            containerFactory = "kafkaManualBatchAckListenerContainerFactory")
    public void test2Listener(List<ConsumerRecord<String, String>> records, Acknowledgment acknowledgment) {
        for (ConsumerRecord<String, String> record : records) {
            log.info("topic.test1 receive : \n" +
                    "data : " + record.value() + "\n" +
                    "key : " + record.key() + "\n" +
                    "topic : " + record.topic() + "\n" +
                    "partitionId : " + record.partition() + "\n" +
                    "offset : " + record.offset() + "\n" +
                    "timestamp : " + record.timestamp() + "\n"
            );
        }
        acknowledgment.acknowledge();
    }
}